using System;
using System.Collections.Generic;
using System.Text;
using Me.Prettyprint.Cassandra.Connection;
using Me.Prettyprint.Cassandra.Model;
using Me.Prettyprint.Cassandra.Model.Thrift;
using Me.Prettyprint.Cassandra.Serializers;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Beans;
using Me.Prettyprint.Hector.Api.Factory;
using Me.Prettyprint.Hector.Api.Query;
using Org.Apache.Cassandra.Thrift;

namespace Me.Prettyprint.Cassandra.Service
{
    /// <summary>
    /// Default implementation of KeyspaceService is used by Nectar to execute
    /// the operations that are supported. The class uses Lambda expressions
    /// and always calls HConnectionManager.operateWithFailover to insure the
    /// operation tries another node if the current one is not responsive.
    /// </summary>
	public class KeyspaceServiceImpl : KeyspaceService
	{
        private static Logger log = new Logger(typeof(KeyspaceServiceImpl));
        private static Dictionary<string, string> EMPTY_CREDENTIALS = new Dictionary<string,string>(0);

        public KeyspaceServiceImpl(String keyspaceName, ConsistencyLevelPolicy consistencyLevel,
            HConnectionManager connectionManager, FailoverPolicy failoverPolicy) :
            this(keyspaceName, consistencyLevel, connectionManager, failoverPolicy, EMPTY_CREDENTIALS)
        {
        }
        public KeyspaceServiceImpl(string keyspaceName, ConsistencyLevelPolicy consistencyLevel,
            HConnectionManager connectionManager, FailoverPolicy failoverPolicy,
            Dictionary<string, string> credentials)
        {
            Consistency = consistencyLevel;
            Name = keyspaceName;
            ConnectionManager = connectionManager;
            FailoverPolicy = failoverPolicy;
            Credentials = credentials;
            ExceptionTranslator = new ExceptionsTranslatorImpl();
        }
        public string Name
        {
            get;
            set;
        }
        protected ConsistencyLevelPolicy Consistency
        {
            get;set;
        }
        protected ExceptionsTranslator ExceptionTranslator
        {
            get;set;
        }
        protected HConnectionManager ConnectionManager
        {
            get;set;
        }
        public CassandraHost CassandraHost
        {
            get;set;
        }
        protected FailoverPolicy FailoverPolicy
        {
            get;set;
        }
        protected Dictionary<string, string> Credentials
        {
            get;set;
        }
        private ConsistencyLevel GetThriftCl(OperationType operationType)
        {
            return ThriftConverter.consistencyLevel(Consistency.Get(operationType));
        }
        private static List<Column> GetColumnList(List<ColumnOrSuperColumn> columns)
        {
            List<Column> list = new List<Column>(columns.Count);
            foreach (ColumnOrSuperColumn col in columns)
            {
                list.Add(col.Column);
            }
            return list;
        }
        private static List<CounterColumn> GetCounterColumnList(List<ColumnOrSuperColumn> columns)
        {
            List<CounterColumn> list = new List<CounterColumn>(columns.Count);
            foreach (ColumnOrSuperColumn col in columns)
            {
                list.Add(col.Counter_column);
            }
            return list;
        }
        private static List<SuperColumn> GetSuperColumnList(List<ColumnOrSuperColumn> columns)
        {
            List<SuperColumn> list = new List<SuperColumn>(columns.Count);
            foreach (ColumnOrSuperColumn col in columns)
            {
                list.Add(col.Super_column);
            }
            return list;
        }
        private static List<CounterSuperColumn> GetCounterSuperColumnList(List<ColumnOrSuperColumn> columns)
        {
            List<CounterSuperColumn> list = new List<CounterSuperColumn>(columns.Count);
            foreach (ColumnOrSuperColumn col in columns)
            {
                list.Add(col.Counter_super_column);
            }
            return list;
        }
        public void BatchMutate(Dictionary<byte[],Dictionary<String,List<Mutation>>> mutationMap)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, string> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    cassandra.batch_mutate(mutationMap, GetThriftCl(OperationType.WRITE));
                    return null;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<string> op = new OperationBuilder<string>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
        }
        public void BatchMutate<K>(BatchMutation<K> batchMutate)
        {
            BatchMutate(batchMutate.ConvertMutationMap());
        }
        public int GetCount(byte[] key, ColumnParent columnParent, SlicePredicate predicate)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, int> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    int count = cassandra.get_count(key, columnParent, predicate, GetThriftCl(OperationType.READ));
                    return count;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<int> op = new OperationBuilder<int>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public Dictionary<byte[], List<Column>> GetRangeSlices(ColumnParent columnParent, 
            SlicePredicate predicate, KeyRange keyRange) 
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Dictionary<byte[], List<Column>>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    List<KeySlice> keySlices = cassandra.get_range_slices(columnParent,
                        predicate, keyRange, GetThriftCl(OperationType.READ));
                    if (keySlices == null || keySlices.Count == 0)
                    {
                        return new Dictionary<byte[], List<Column>>(0);
                    }
                    Dictionary<byte[], List<Column>> ret = new Dictionary<byte[], List<Column>>(keySlices.Count);
                    foreach (KeySlice keySlice in keySlices)
                    {
                        ret[keySlice.Key] = GetColumnList(keySlice.Columns);
                    }
                    return ret;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<Dictionary<byte[], List<Column>>> op = new OperationBuilder<Dictionary<byte[], List<Column>>>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public Dictionary<byte[], List<CounterColumn>> GetRangeCounterSlices(ColumnParent columnParent,
            SlicePredicate predicate, KeyRange keyRange)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Dictionary<byte[], List<CounterColumn>>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    List<KeySlice> keySlices = cassandra.get_range_slices(columnParent,
                        predicate, keyRange, GetThriftCl(OperationType.READ));
                    if (keySlices == null || keySlices.Count == 0)
                    {
                        return new Dictionary<byte[], List<CounterColumn>>(0);
                    }
                    Dictionary<byte[], List<CounterColumn>> ret = new Dictionary<byte[], List<CounterColumn>>(keySlices.Count);
                    foreach (KeySlice keySlice in keySlices)
                    {
                        ret[keySlice.Key] = GetCounterColumnList(keySlice.Columns);
                    }
                    return ret;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<Dictionary<byte[], List<CounterColumn>>> op = new OperationBuilder<Dictionary<byte[], List<CounterColumn>>>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public Dictionary<byte[], List<SuperColumn>> GetSuperRangeSlices(ColumnParent columnParent,
            SlicePredicate predicate, KeyRange keyRange)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Dictionary<byte[], List<SuperColumn>>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    List<KeySlice> keySlices = cassandra.get_range_slices(columnParent,
                        predicate, keyRange, GetThriftCl(OperationType.READ));
                    if (keySlices == null || keySlices.Count == 0)
                    {
                        return new Dictionary<byte[], List<SuperColumn>>(0);
                    }
                    Dictionary<byte[], List<SuperColumn>> ret = new Dictionary<byte[], List<SuperColumn>>(keySlices.Count);
                    foreach (KeySlice keySlice in keySlices)
                    {
                        ret[keySlice.Key] = GetSuperColumnList(keySlice.Columns);
                    }
                    return ret;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<Dictionary<byte[], List<SuperColumn>>> op = new OperationBuilder<Dictionary<byte[], List<SuperColumn>>>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public Dictionary<byte[], List<CounterSuperColumn>> GetSuperRangeCounterSlices(ColumnParent columnParent,
            SlicePredicate predicate, KeyRange keyRange)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Dictionary<byte[], List<CounterSuperColumn>>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    List<KeySlice> keySlices = cassandra.get_range_slices(columnParent,
                        predicate, keyRange, GetThriftCl(OperationType.READ));
                    if (keySlices == null || keySlices.Count == 0)
                    {
                        return new Dictionary<byte[], List<CounterSuperColumn>>(0);
                    }
                    Dictionary<byte[], List<CounterSuperColumn>> ret = new Dictionary<byte[], List<CounterSuperColumn>>(keySlices.Count);
                    foreach (KeySlice keySlice in keySlices)
                    {
                        ret[keySlice.Key] = GetCounterSuperColumnList(keySlice.Columns);
                    }
                    return ret;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<Dictionary<byte[], List<CounterSuperColumn>>> op = new OperationBuilder<Dictionary<byte[], List<CounterSuperColumn>>>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public List<Column> GetSlice(byte[] key, ColumnParent columnParent,
            SlicePredicate predicate) 
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, List<Column>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    List<ColumnOrSuperColumn> cosclist = cassandra.get_slice(key, columnParent,
                        predicate, GetThriftCl(OperationType.READ));
                    if (cosclist == null)
                    {
                        return null;
                    }
                    List<Column> result = new List<Column>(cosclist.Count);
                    foreach (ColumnOrSuperColumn cosc in cosclist)
                    {
                        result.Add(cosc.Column);
                    }
                    return result;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<List<Column>> op = new OperationBuilder<List<Column>>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public List<Column> GetSlice(string key, ColumnParent columnParent, SlicePredicate predicate)
        {
            return GetSlice(StringSerializer.Instance.ToBytes(key), columnParent, predicate);
        }
        public List<CounterColumn> GetCounterSlice(byte[] key, ColumnParent columnParent,
            SlicePredicate predicate)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, List<CounterColumn>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    List<ColumnOrSuperColumn> cosclist = cassandra.get_slice(key, columnParent,
                        predicate, GetThriftCl(OperationType.READ));
                    if (cosclist == null)
                    {
                        return null;
                    }
                    List<CounterColumn> result = new List<CounterColumn>(cosclist.Count);
                    foreach (ColumnOrSuperColumn cosc in cosclist)
                    {
                        result.Add(cosc.Counter_column);
                    }
                    return result;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<List<CounterColumn>> op = new OperationBuilder<List<CounterColumn>>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public List<CounterColumn> GetCounterSlice(string key, ColumnParent columnParent, SlicePredicate predicate)
        {
            return GetCounterSlice(StringSerializer.Instance.ToBytes(key), columnParent, predicate);
        }
        public SuperColumn GetSuperColumn(byte[] key, ColumnPath columnPath)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, SuperColumn> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    ColumnOrSuperColumn cosc = cassandra.get(key, columnPath, GetThriftCl(OperationType.READ));
                    return cosc == null ? null : cosc.Super_column;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<SuperColumn> op = new OperationBuilder<SuperColumn>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public List<SuperColumn> GetSuperSlice(string key, ColumnParent columnParent, SlicePredicate predicate) 
        {
            return GetSuperSlice(StringSerializer.Instance.ToBytes(key), columnParent, predicate);
        }
        public List<SuperColumn> GetSuperSlice(byte[] key, ColumnParent columnParent, SlicePredicate predicate)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, List<SuperColumn>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    List<ColumnOrSuperColumn> cosclist = cassandra.get_slice(key, columnParent,
                        predicate, GetThriftCl(OperationType.READ));
                    if (cosclist == null) 
                    {
                        return null;
                    }
                    List<SuperColumn> result = new List<SuperColumn>(cosclist.Count);
                    foreach (ColumnOrSuperColumn cosc in cosclist)
                    {
                        result.Add(cosc.Super_column);
                    }
                    return result;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<List<SuperColumn>> op = new OperationBuilder<List<SuperColumn>>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public SuperColumn GetSuperColumn(byte[] key, ColumnPath columnPath,
            bool reversed, int size)
        {
            SliceRange sliceRange = new SliceRange(new byte[0],new byte[0], reversed, size);
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, SuperColumn> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    ColumnParent clp = new ColumnParent(columnPath.Column_family);
                    clp.Super_column = columnPath.Super_column;
                    SlicePredicate sp = new SlicePredicate();
                    sp.Slice_range = sliceRange;
                    List<ColumnOrSuperColumn> cosc = cassandra.get_slice(key, clp, sp,
                        GetThriftCl(OperationType.READ));
                    if (cosc == null || cosc.Count == 0)
                    {
                        return null;
                    }
                    return new SuperColumn(columnPath.Super_column, GetColumnList(cosc));
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<SuperColumn> op = new OperationBuilder<SuperColumn>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public SuperColumn GetSuperColumn(string key, ColumnPath columnPath)
        {
            return GetSuperColumn(StringSerializer.Instance.ToBytes(key), columnPath);
        }
        public List<CounterSuperColumn> GetCounterSuperSlice(string key, ColumnParent columnParent, SlicePredicate predicate)
        {
            return GetCounterSuperSlice(StringSerializer.Instance.ToBytes(key), columnParent, predicate);
        }
        public List<CounterSuperColumn> GetCounterSuperSlice(byte[] key, ColumnParent columnParent,
            SlicePredicate predicate)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, List<CounterSuperColumn>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    List<ColumnOrSuperColumn> cosclist = cassandra.get_slice(key, columnParent,
                        predicate, GetThriftCl(OperationType.READ));
                    if (cosclist == null)
                    {
                        return null;
                    }
                    List<CounterSuperColumn> result = new List<CounterSuperColumn>(cosclist.Count);
                    foreach (ColumnOrSuperColumn cosc in cosclist)
                    {
                        result.Add(cosc.Counter_super_column);
                    }
                    return result;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<List<CounterSuperColumn>> op = new OperationBuilder<List<CounterSuperColumn>>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public void Insert(string key, ColumnParent columnParent, Column column)
        {
            Insert(StringSerializer.Instance.ToBytes(key), columnParent, column);
        }
        public void Insert(byte[] key, ColumnParent columnParent, Column column) 
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, string> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    cassandra.insert(key, columnParent, column, GetThriftCl(OperationType.WRITE));
                    return null;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<string> op = new OperationBuilder<string>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
        }
        public void AddCounter(byte[] key, ColumnParent columnParent, CounterColumn counterColumn)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, string> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    cassandra.add(key, columnParent, counterColumn, GetThriftCl(OperationType.WRITE));
                    return null;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<string> op = new OperationBuilder<string>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
        }
        public void AddCounter(string key, ColumnParent columnParent, CounterColumn counterColumn)
        {
            AddCounter(StringSerializer.Instance.ToBytes(key), columnParent, counterColumn);
        }
        public void Insert(string key, ColumnPath columnPath, byte[] value) 
        {
            ColumnParent columnParent = new ColumnParent(columnPath.Column_family);
            if (columnPath.__isset.super_column)
            {
                columnParent.Super_column = columnPath.Super_column;
            }
            Column column = new Column(columnPath.Column);
            column.Value = value;
            column.Timestamp = ConnectionManager.CreateClock();
            Insert(StringSerializer.Instance.ToBytes(key), columnParent, column);
        }
        public void Insert(string key, ColumnPath columnPath, byte[] value, long timestamp) 
        {
            ColumnParent columnParent = new ColumnParent(columnPath.Column_family);
            if (columnPath.__isset.super_column)
            {
                columnParent.Super_column = columnPath.Super_column;
            }
            Column column = new Column(columnPath.Column);
            column.Value = value;
            column.Timestamp = timestamp;
            Insert(StringSerializer.Instance.ToBytes(key), columnParent, column);
        }
        public Dictionary<byte[], List<Column>> MultigetSlice(List<byte[]> keys,
            ColumnParent columnParent, SlicePredicate predicate) 
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Dictionary<byte[], List<Column>>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    Dictionary<byte[], List<ColumnOrSuperColumn>> cfmap = cassandra.multiget_slice(
                        keys, columnParent, predicate, GetThriftCl(OperationType.READ));
                    Dictionary<byte[], List<Column>> result = new Dictionary<byte[], List<Column>>();
                    foreach (byte[] key in cfmap.Keys)
                    {
                        result[key] = GetColumnList(cfmap[key]);
                    }
                    return result;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<Dictionary<byte[], List<Column>>> op = new OperationBuilder<Dictionary<byte[], List<Column>>>(x);
            op.OperationType = OperationType.READ;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public Dictionary<byte[], List<CounterColumn>> MultigetCounterSlice(List<byte[]> keys,
            ColumnParent columnParent, SlicePredicate predicate)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Dictionary<byte[], List<CounterColumn>>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    Dictionary<byte[], List<ColumnOrSuperColumn>> cfmap = cassandra.multiget_slice(
                        keys, columnParent, predicate, GetThriftCl(OperationType.READ));
                    Dictionary<byte[], List<CounterColumn>> result = new Dictionary<byte[], List<CounterColumn>>();
                    foreach (byte[] key in cfmap.Keys)
                    {
                        result[key] = GetCounterColumnList(cfmap[key]);
                    }
                    return result;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<Dictionary<byte[], List<CounterColumn>>> op = new OperationBuilder<Dictionary<byte[], List<CounterColumn>>>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public Dictionary<byte[], SuperColumn> MultigetSuperColumn(List<byte[]> keys, ColumnPath columnPath)
        {
            return MultigetSuperColumn(keys, columnPath, false, int.MaxValue);
        }
        public Dictionary<byte[], SuperColumn> MultigetSuperColumn(List<byte[]> keys, ColumnPath columnPath,
            bool reversed, int size)
        {
            ColumnParent clp = new ColumnParent(columnPath.Column_family);
            clp.Super_column = columnPath.Super_column;
            
            SliceRange sr = new SliceRange(new byte[0], new byte[0], reversed, size);
            SlicePredicate sp = new SlicePredicate();
            sp.Slice_range = sr;
            
            Dictionary<byte[], List<SuperColumn>> sclist = MultigetSuperSlice(keys, clp, sp);
            
            if (sclist == null || sclist.Count == 0)
            {
                return new Dictionary<byte[],SuperColumn>();
            }
            
            Dictionary<byte[], SuperColumn> result = new Dictionary<byte[], SuperColumn>(keys.Count * 2);
            foreach (byte[] key in sclist.Keys)
            {
                List<SuperColumn> sclistByKey = sclist[key];
                if (sclistByKey.Count > 0)
                {
                    result[key] = sclistByKey[0];
                }
            }
            return result;
        }
        public Dictionary<byte[], List<SuperColumn>> MultigetSuperSlice(List<byte[]> keys,
            ColumnParent columnParent, SlicePredicate predicate)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Dictionary<byte[], List<SuperColumn>>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    Dictionary<byte[], List<ColumnOrSuperColumn>> cfmap = cassandra.multiget_slice(
              keys, columnParent, predicate, GetThriftCl(OperationType.READ));
                    if (!columnParent.__isset.super_column)
                    {
                        Dictionary<byte[], List<SuperColumn>> result = new Dictionary<byte[], List<SuperColumn>>();
                        foreach (byte[] key in cfmap.Keys)
                        {
                            result[key] = GetSuperColumnList(cfmap[key]);
                        }
                        return result;
                    }
                    else
                    {
                        Dictionary<byte[], List<SuperColumn>> result = new Dictionary<byte[], List<SuperColumn>>();
                        foreach (byte[] key in cfmap.Keys)
                        {
                            SuperColumn spc = new SuperColumn(columnParent.Super_column, GetColumnList(cfmap[key]));
                            List<SuperColumn> spclist = new List<SuperColumn>(1);
                            spclist.Add(spc);
                            result[key] = spclist;
                        }
                        return result;
                    }
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<Dictionary<byte[], List<SuperColumn>>> op = new OperationBuilder<Dictionary<byte[], List<SuperColumn>>>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public Dictionary<byte[], List<CounterSuperColumn>> MultigetCounterSuperSlice(List<byte[]> keys,
            ColumnParent columnParent, SlicePredicate predicate)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Dictionary<byte[], List<CounterSuperColumn>>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    Dictionary<byte[], List<ColumnOrSuperColumn>> cfmap = cassandra.multiget_slice(
                        keys, columnParent, predicate, GetThriftCl(OperationType.READ));
                    if (!columnParent.__isset.super_column)
                    {
                        Dictionary<byte[], List<CounterSuperColumn>> result = new Dictionary<byte[], List<CounterSuperColumn>>();
                        foreach (byte[] key in cfmap.Keys)
                        {
                            result[key] =  GetCounterSuperColumnList(cfmap[key]);
                        }
                        return result;
                    }
                    else
                    {
                        Dictionary<byte[], List<CounterSuperColumn>> result = new Dictionary<byte[], List<CounterSuperColumn>>();
                        foreach (byte[] key in cfmap.Keys)
                        {
                            CounterSuperColumn spc = new CounterSuperColumn(columnParent.Super_column,
                                GetCounterColumnList(cfmap[key]));
                            List<CounterSuperColumn> spclist = new List<CounterSuperColumn>(1);
                            spclist.Add(spc);
                            result[key] = spclist;
                        }
                        return result;
                    }
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<Dictionary<byte[], List<CounterSuperColumn>>> op = new OperationBuilder<Dictionary<byte[], List<CounterSuperColumn>>>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public Dictionary<byte[], List<Column>> GetIndexedSlices(ColumnParent columnParent,
            IndexClause indexClause, SlicePredicate predicate)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Dictionary<byte[], List<Column>>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    List<KeySlice> keySlices = cassandra.get_indexed_slices(columnParent, indexClause,
                        predicate, GetThriftCl(OperationType.READ));
                    if (keySlices == null || keySlices.Count == 0) 
                    {
                        return new Dictionary<byte[], List<Column>>(0);
                    }
                    Dictionary<byte[], List<Column>> ret = new Dictionary<byte[], List<Column>>(keySlices.Count);
                    foreach (KeySlice keySlice in keySlices)
                    {
                        ret[keySlice.Key] = GetColumnList(keySlice.Columns);
                    }
                    return ret;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<Dictionary<byte[], List<Column>>> op = new OperationBuilder<Dictionary<byte[], List<Column>>>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public void Remove(byte[] key, ColumnPath columnPath)
        {
            Remove(key, columnPath, ConnectionManager.CreateClock());
        }
        public void Remove(byte[] key, ColumnPath columnPath, long timestamp)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, string> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    cassandra.remove(key, columnPath, timestamp, GetThriftCl(OperationType.WRITE));
                    return null;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<string> op = new OperationBuilder<string>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
        }
        public Dictionary<byte[], int> MultigetCount(List<byte[]> keys,
            ColumnParent columnParent, SlicePredicate slicePredicate)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Dictionary<byte[], int>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    Dictionary<byte[], int> ret = cassandra.multiget_count(keys, columnParent, slicePredicate,
                        GetThriftCl(OperationType.READ));
                    return ret;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<Dictionary<byte[], int>> op = new OperationBuilder<Dictionary<byte[], int>>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public void RemoveCounter(byte[] key, ColumnPath columnPath)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, string> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    cassandra.remove_counter(key, columnPath, GetThriftCl(OperationType.WRITE));
                    return null;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<string> op = new OperationBuilder<string>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
        }
        public void RemoveCounter(string key, ColumnPath columnPath) 
        {
            RemoveCounter(StringSerializer.Instance.ToBytes(key), columnPath);
        }
        public void Remove(string key, ColumnPath columnPath)
        {
            Remove(StringSerializer.Instance.ToBytes(key), columnPath);
        }
        public void Remove(string key, ColumnPath columnPath, long timestamp)
        {
            Remove(StringSerializer.Instance.ToBytes(key), columnPath, timestamp);
        }
        public Column GetColumn(byte[] key, ColumnPath columnPath)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Column> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    ColumnOrSuperColumn cosc = cassandra.get(key, columnPath, GetThriftCl(OperationType.READ));
                    return cosc == null ? null : cosc.Column;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<Column> op = new OperationBuilder<Column>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public CounterColumn GetCounter(byte[] key, ColumnPath columnPath)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, CounterColumn> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    ColumnOrSuperColumn cosc = cassandra.get(key, columnPath, GetThriftCl(OperationType.READ));
                    return cosc == null ? null : cosc.Counter_column;
                }
                catch (Exception e)
                {
                    throw ExceptionTranslator.translate(e);
                }
            };
            Operation<CounterColumn> op = new OperationBuilder<CounterColumn>(x);
            op.OperationType = OperationType.WRITE;
            op.FailoverPolicy = FailoverPolicy;
            op.KeyspaceName = Name;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            CassandraHost = op.CassandraHost;
            return op.Result;
        }
        public CounterColumn GetCounter(string key, ColumnPath columnPath)
        {
            return GetCounter(StringSerializer.Instance.ToBytes(key), columnPath);
        }
        public Column GetColumn(string key, ColumnPath columnPath)
        {
            return GetColumn(StringSerializer.Instance.ToBytes(key), columnPath);
        }
        public HConsistencyLevel GetConsistencyLevel(OperationType operationType)
        {
            return Consistency.Get(operationType);
        }
        public override string ToString()
        {
            StringBuilder b = new StringBuilder();
            b.Append("KeyspaceImpl<");
            b.Append(Name);
            b.Append(">");
            return b.ToString();
        }
	}
}